package com.rabbitmq.example.idempotent;


import com.rabbitmq.client.Channel;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.PublisherCallbackChannel;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;

import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.util.concurrent.TimeUnit;

@Aspect
public class IdempotentConsumerAspect {
    private static Logger logger = LoggerFactory.getLogger(IdempotentConsumerAspect.class);

    private ValueOperations<String, String> valueOps;

    public IdempotentConsumerAspect(StringRedisTemplate stringRedisTemplate) {
        valueOps = stringRedisTemplate.opsForValue();
    }

    @Pointcut("@annotation(com.rabbitmq.example.idempotent.IdempotentConsumerAn)")
    public void IdempotentPointCut() {}

    @Around("IdempotentPointCut()")
    public Object around(ProceedingJoinPoint point) throws Throwable{
        logger.info("consumer now in Aspect");

        Signature signature = point.getSignature();
        MethodSignature methodSignature = (MethodSignature) signature;
        Method method = methodSignature.getMethod();
        Parameter[] parameters = method.getParameters();
        Object[] args = point.getArgs();
        IdempotentConsumerAn annotation = method.getAnnotation(IdempotentConsumerAn.class);
        if (annotation == null) {
            logger.info("this method not using IdempotentConsumerAn annotation");
            return point.proceed();
        }else {
            logger.info("this method using IdempotentConsumerAn annotation");
            Message message = null;
            Channel channel = null;

            for (Object arg : args) {
                if(arg instanceof org.springframework.amqp.core.Message) {
                    message = (org.springframework.amqp.core.Message) arg;
                }
                if(arg instanceof Channel) {
                    channel = (Channel) arg;
                }
            }

            if(message == null || channel == null) {
                logger.error("message and channel must not null");
                throw new Exception("message and channel must not null");
            }

            //消息的幂等性可以借助  spring_returned_message_correlation   header 中 的id 进行确认,该值是我们自己设置的

            String correlation = message.getMessageProperties().getHeader(PublisherCallbackChannel.RETURNED_MESSAGE_CORRELATION_KEY);

            //correlation  可以唯一确定message    ack  超时的话，消息会重回队列  非ack 的情况下  是在 noack 的状态而不是 read 的状态
            //redis  判断是否存在这个id   不存在 进行
            // if(不存在)
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            Boolean isOk = valueOps.setIfAbsent(correlation, "1", 30, TimeUnit.DAYS);
            try {
                if(isOk) {
                    logger.info("this message not using  so consuming,deliveryTag {}",deliveryTag);
                    return point.proceed();
                }
            }finally {
                channel.basicAck(deliveryTag,true);
            }
        }

        logger.info("this message used  so return");
        return null;
    }
}
